package com.colorlight.cloud.mqtt.base.core.handler;

import com.colorlight.cloud.mqtt.base.api.message.ProcessMessage;
import com.colorlight.cloud.mqtt.base.core.pool.ThreadPool;
import com.colorlight.cloud.mqtt.base.entity.SubMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.*;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;


/**
 * @author Demon
 * 消息处理
 */
@Component("MessageHandler")
@Slf4j
public class MqttMessageHandler implements MessageHandler {
	
	
	private final ProcessMessage processMessage;
	private final ThreadPool threadPool;
	
	public MqttMessageHandler(ProcessMessage processMessage, ThreadPool threadPool) {
		this.processMessage = processMessage;
		this.threadPool = threadPool;
	}
	
	@Override
	public void handleMessage(Message<?> message) throws MessagingException {
		//加入线程池  分流消息  防止消息堵塞
		threadPool.subscriptionEventDeliverTaskExecutor().execute(() -> {
			individualMessageProcessing(message);
		});
	}
	
	private void individualMessageProcessing(Message<?> message) {
		//消息上报 主题中带有消息类型  ${env_Prefix}-update/${productKey}/$Name/$DeviceName
		MessageHeaders headers = message.getHeaders();
		String topic = (String) headers.get(MqttHeaders.RECEIVED_TOPIC);
		String payload = (String) message.getPayload();
		byte[] bytes = payload.getBytes(StandardCharsets.ISO_8859_1);
		SubMessage<byte[]> subMessage = new SubMessage<>();
		subMessage.setMessage(bytes);
		subMessage.setTopic(topic);
		subscriptionEventCreate(subMessage);
	}
	
	public void subscriptionEventCreate(SubMessage<byte[]> subMessage) {
		try {
			log.debug("消息上报\n 上报主题:{} \n消息详情:{}", subMessage.getTopic(), subMessage.getMessage());
			processMessage.process(subMessage);
		} catch (Exception e) {
			log.error("消息上报处理异常\n 上报主题:{} \n消息详情:{}\n异常详情:", subMessage.getTopic(), subMessage.getMessage(), e);
		}
	}
}